package com.thrift.client.utils;

import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.retry.RetryForever;
import org.apache.curator.utils.CloseableUtils;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.ACL;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Configuration;

import java.util.List;

/**
 * @author 何锦涛--CodeSleep
 * @since 2022/9/11 10:11
 */
@Data
@Configuration
@NoArgsConstructor
public class ZkUtil implements AutoCloseable{

    private static final Logger logger = LoggerFactory.getLogger(ZkUtil.class);

    // zk 中的服务名称
    private static String name;

    // zk 服务器连接字符串
    @Value("${spring.cloud.zookeeper.connect-string}")
    private static String zkConnectedStr;

    // Curator 框架封装好的 Client
    private static CuratorFramework zkClient;

    /**
     * 自定义线程异常捕获器，这里只是打印暂无其他操作
     */
    private static Thread.UncaughtExceptionHandler uncaughtExceptionHandler = new Thread.UncaughtExceptionHandler() {
        @Override
        public void uncaughtException(Thread t, Throwable e) {
            logger.error(name + " 发生线程异常：" + e);
        }
    };


    /**
     * 构造函数，启动 zk 客户端连接 zk 服务器
     * @param zkConnectedStr 服务器连接字符串
     */
    public ZkUtil(String zkConnectedStr) {
        try {
            this.zkConnectedStr = zkConnectedStr;
            RetryPolicy retryPolicy = new RetryForever(10000);
            this.zkClient = CuratorFrameworkFactory.builder()
                    .connectString(zkConnectedStr)      // 指 zk server 列表，由英文逗号分开的 host:port 组成，就是集群地址
                    .retryPolicy(retryPolicy)           // 失败重试策略
                    .sessionTimeoutMs(30 * 1000)        // 会话超时时间，单位为 ms
                    .connectionTimeoutMs(30 * 1000)     // 连接创建超时时间，单位为 ms
                    .maxCloseWaitMs(60 * 1000)          // 关闭加入后台线程时的等待时间
                    .build();
            this.zkClient.start();
            this.zkClient.blockUntilConnected();
            logger.info(String.format("TserverZK 启动，zk 状态为"
                    + ZooKeeper.States.CONNECTED + " Curator 状态为 "
                    + zkClient.getState()
                    + " :  %s started.", this.zkConnectedStr));
        } catch (InterruptedException e) {
            logger.error("构造函数，启动 zk 客户端连接 zk 服务器失败：{}", e.toString());
        }
    }

    /**
     * 单机模式自动装配
     */
    public static CuratorFramework zookeeperClient() {
        try {
            RetryPolicy retryPolicy = new RetryForever(10000);
            zkClient = CuratorFrameworkFactory.builder()
                    .connectString("192.168.217.3:2181, 192.168.217.3:2182, 192.168.217.3:2183")      // 指 zk server 列表，由英文逗号分开的 host:port 组成，就是集群地址
                    .retryPolicy(retryPolicy)           // 失败重试策略
                    .sessionTimeoutMs(30 * 1000)        // 会话超时时间，单位为 ms
                    .connectionTimeoutMs(30 * 1000)     // 连接创建超时时间，单位为 ms
                    .maxCloseWaitMs(60 * 1000)          // 关闭加入后台线程时的等待时间
                    .build();
            zkClient.start();
            zkClient.blockUntilConnected();
            logger.info(String.format("TserverZK 启动，zk 状态为"
                    + ZooKeeper.States.CONNECTED + " Curator 状态为 "
                    + zkClient.getState()
                    + " :  %s started.", zkConnectedStr));
        } catch (InterruptedException e) {
            logger.error("构造函数，启动 zk 客户端连接 zk 服务器失败：{}", e.toString());
        }
        return zkClient;
    }

    /**
     * 添加节点，指定节点类型
     * @param path 创建路径
     * @param value 数据内容
     * @param mode 节点类型，不加 .withMode 默认为持久性节点
     *             CreateMode.PERSISTENT：持久性节点
     */
    public static String add(String path, byte[] value, CreateMode mode) {
        try {
            // creatingParentContainersIfNeeded：如果尚未创建任何父节点，则使用 CreateMode.CONTAINER 创建它们。
            return zkClient.create().creatingParentContainersIfNeeded()
                    .withMode(mode).forPath(path, value);
        } catch (Exception e) {
            logger.error("zk 添加节点时发生异常：{}", e.toString());
            throw new RuntimeException(e);
        }
    }

    /**
     * 添加节点，指定节点类型，并且需要认证
     * @param path 创建路径
     * @param value 数据内容
     * @param mode 节点类型，不加 .withMode 默认为持久性节点
     * @param aclList 要使用的 ACL 认证列表
     */
    public static String add(String path, byte[] value, CreateMode mode, List<ACL> aclList) {
        try {
            return zkClient.create().creatingParentContainersIfNeeded()
                    .withMode(mode).withACL(aclList).forPath(path, value);
        } catch (Exception e) {
            logger.error("zk 添加 ACL 节点时发生异常：{}", e.toString());
            throw new RuntimeException(e);
        }
    }

    /**
     * 判断节点是否存在
     * @param path 节点路径
     * @return 是否存在
     */
    public static boolean exist(String path) {
        try {
            return zkClient.checkExists().forPath(path) != null;
        } catch (Exception e) {
            logger.error("判断节点 path {} 是否存在发生异常：{}", path, e.toString());
            return false;
        }
    }

    /**
     * 移除节点
     * @param path 节点路径
     */
    public static void remove(String path) {
        try {
            zkClient.delete().forPath(path);
        } catch (Exception e) {
            logger.error("zk 移除节点 path {} 时发生异常：{}", path, e.toString());
            throw new RuntimeException(e);
        }
    }

    /**
     * 获取节点下的所有子节点
     * @param nodePath 节点
     * @return 子节点
     */
    public static List<String> getChildren(String nodePath) {
        try {
            return zkClient.getChildren().forPath(nodePath);
        } catch (Exception e) {
            logger.error("zk 获取节点 nodePath {} 下的所有节点时发生异常：{}", nodePath, e.toString());
        }
        return null;
    }

    /**
     * 注册目录监听器
     * @param nodePath
     * @param listener
     * @return
     */
    public static PathChildrenCache registerPathChildrenListener(String nodePath, PathChildrenCacheListener listener) {
        try {
            PathChildrenCache pathChildrenCache = new PathChildrenCache(zkClient, nodePath, true);
            // 添加子目录监视器
            pathChildrenCache.getListenable().addListener(listener);
            // 启动监听器
            pathChildrenCache.start(PathChildrenCache.StartMode.BUILD_INITIAL_CACHE);
            return pathChildrenCache;
        } catch (Exception e) {
            logger.error("zk 注册节点 nodePath {} 目录监听器时发生异常：{}", nodePath, e.toString());
        }
        return null;
    }

    @Override
    public void close() throws Exception {
        logger.info(String.format("TserverZK 关闭 zkname： %s --- zkConnectedStr： %s closed"
                , this.name, this.zkConnectedStr));
        CloseableUtils.closeQuietly(this.zkClient);
    }
}